package com.neusoft.databus.core.service.impl;

import java.util.Date;

import javax.annotation.PostConstruct;

import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.Subscribe;
import com.neusoft.bizcore.common.bean.MessageBean;
import com.neusoft.bizcore.common.messaging.ProducerService;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.core.service.MonitorDataDispatcherService;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class MonitorDataDispatcherServiceIml implements MonitorDataDispatcherService {

    @javax.annotation.Resource(name = "monitorDataEventBus")
    private AsyncEventBus monitorDataEventBus;
    @Value("${databus.core.policy.monitorDataTopic: ''}")
    private String policyTopic;
    @Autowired
    private ProducerService producerService;

    @PostConstruct
    public void init() {
        this.monitorDataEventBus.register(this);
    }

    @Override
    @AllowConcurrentEvents // 线程安全
    @Subscribe
    public void dispatcher(final CommandBean commandBean) {
        try {
            MDC.put("trace", commandBean.generateTraceId());
            if (MonitorDataDispatcherServiceIml.log.isInfoEnabled()) {
                MonitorDataDispatcherServiceIml.log.info(
                        "monitor-data-eventbus: ready to dispatch monitor data to messaging middleware: {}",
                        commandBean.generateTraceId());
            }
            final Date now = new Date();
            commandBean.clearNotCoreFileds();
            commandBean.setCollectEndTime(now);
            final MessageBean message =
                    new MessageBean(JsonUtils.pojoToJsonWithoutNullFields(commandBean).toString(), now);
            if (!"".equals(this.policyTopic)) {
                this.producerService.send(this.policyTopic, message);
            }
            MDC.clear();
        } catch (final Exception e) {
            MonitorDataDispatcherServiceIml.log.error(e.getMessage(), "");
        }
    }

}
